package com.mtiiot.fstcpserver.mq;

import com.mtiiot.fstcpserver.base.BaseService;
import com.mtiiot.fstcpserver.mq.kafka.insplatform.InstruProducer;
import com.mtiiot.fstcpserver.utils.Constants;
import com.mtiiot.fscore.bean.MetadataInfo;
import com.mtiiot.fscore.bean.MqInfo;
import com.mtiiot.fscore.bean.elasticsearch.metadata.RedisMqMetadata;
import com.mtiiot.fscore.inter1.KafkaMqService1;
import com.mtiiot.fscore.utils.RequestType;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.codehaus.jackson.map.ObjectMapper;
import org.springframework.transaction.annotation.Transactional;

/**
 * Created by Administrator on 16-7-22.
 */

@Transactional
public class KafkaMqServiceImpl1 extends BaseService implements KafkaMqService1 {

    private static final Logger logger = LogManager.getLogger(KafkaMqServiceImpl1.class);

    /**
     * 生产单条消息
     *
     * @param metadataInfo 元数据信息
     * @param mqInfo
     * @return int 成功的消息条数
     */
    @Override
    public int produceMessage(MetadataInfo metadataInfo, MqInfo mqInfo) {
        try{
            //1  异步数据入kafka队列
            checkMetadata(metadataInfo);//验证元数据
            ObjectMapper mapper = new ObjectMapper();
            String json=mapper.writeValueAsString(mqInfo);
            String topic = mqInfo.getTopic();
            String key = mqInfo.getKey();
            String msg =(String) mqInfo.getValue();
            InstruProducer.sendData(topic,key,msg);


            //2  元数据入reids队列
            RedisMqMetadata md=new RedisMqMetadata();
            setMetadataBeanData(metadataInfo,md, RequestType.INSERT.getName(),json);
            md.setMqTopic(mqInfo.getTopic());
            md.setMqKey(mqInfo.getKey());
            md.setMqValue(mqInfo.getValue());
            metadataInQueue(Constants.TOPIC_METADATAINFO,Constants.TYPE_METADATA_KAFKAMQ,md);
            return 1;
        }catch (Exception ex){
            //3 异常信息 元数据入reids队列
            handleException(Constants.MODULE_MQ_REDIS,"RedisMqServiceImpl1","produceMessage",ex,logger);
            return 0;
        }
    }

}
